package com.huitone.smspfm.task;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.huitone.smspfm.logger.ExtLogger;
import com.huitone.smspfm.model.SmsEntityForMt;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.constant.DealResult;
import com.huitone.smspfm.core.filter.ForbidTimeFilterTask;
import com.huitone.smspfm.dao.oracle.entity.GateDef;
/**   
* @Description: 缓存队列处理任务
* @author ozh   
* @date 2019年11月05日 下午09:52:28 
* @version V1.0   
*/
public class CacheQueueProcessingTask extends Thread {
	
	private static ExtLogger logger = ExtLogger.create(CacheQueueProcessingTask.class);
	
	private ExecutorService blockingPool = new ThreadPoolExecutor(AppStaticConfigs.cacheQueueExcutePoolSize, 
			AppStaticConfigs.cacheQueueExcutePoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.cacheQueueExcutePoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	
	private boolean toSleep = false;
	//间隔1分钟
	private long sleepTime = 1*60*1000;
	
	
	private boolean keepGoing = true;
	

	private static RedisScript<List> luaGetCacheListNameScript = null;
	
	
	private static RedisScript<List> cacheQueueLengthScript = null;
	

	RedisScript<List> luaGetCacheQueueSmsScript = null;
	
	@Override
	public void run() {
		
		luaGetCacheListNameScript = new DefaultRedisScript<>(AppStaticConfigs.luaGetCacheListName, List.class);			
		cacheQueueLengthScript = new DefaultRedisScript<List>(AppStaticConfigs.luaGetCacheQueueLength, List.class);
		luaGetCacheQueueSmsScript = new DefaultRedisScript<List>(AppStaticConfigs.luaGetCacheQueueSms, List.class);
		
				
		List<String> luaGetCacheListNameKeys = new ArrayList<>();
		luaGetCacheListNameKeys.add(AppStaticConfigs.isCacheQueueListKey);
		luaGetCacheListNameKeys.add(StringUtils.toString(AppStaticConfigs.isCacheQueueListQuerySize));
		
		List<String> cacheQueueLengthKeys = new ArrayList<>();
		cacheQueueLengthKeys.add(AppStaticConfigs.isCacheQueueListKey);	
		
		List<String> priQueueLengthKeys = new ArrayList<>();
		priQueueLengthKeys.add(AppStaticConfigs.iSMPriListNamesSet);
		
		boolean isRunning = false;
		
		while(isKeepGoing() && !this.isInterrupted()) {
			
			Long startTime = System.nanoTime();
			logger.info("start 缓存队列处理任务");
			if(!isRunning){
				isRunning = true;
				try {
				rootHandler(luaGetCacheListNameKeys, cacheQueueLengthKeys,
					priQueueLengthKeys);
				} catch (Exception ex){
					ex.printStackTrace();
					logger.error("处理缓存短信队列时 异常 ",ex);
				}
				isRunning = false;
			}
			// 无论如何都要等待间隔
			toSleep = true;
			Long endTime = System.nanoTime();
			logger.info("缓存队列处理消耗时间["+(endTime-startTime)+"]");
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {}
		}
		
		if (blockingPool != null && !blockingPool.isShutdown()) {
			blockingPool.shutdown();
		}		
		
		logger.info("短信平台缓存队列处理服务已经关闭");
	}


	private void rootHandler(List<String> luaGetCacheListNameKeys,
			List<String> cacheQueueLengthKeys, List<String> priQueueLengthKeys) {
		//所有缓存队列
		List<String> cacheQueueNameList = null;
		List<Long> cacheQueueLengthList = null;
		List<Long> priQueueLengthList = null;
		int handlerNum = 0;
		List<String> luaGetCacheListNameArgs = new ArrayList<String>();
		luaGetCacheListNameArgs.add("1000");
		try {
			cacheQueueNameList = AppStaticConfigs.scriptOpts3.getList(luaGetCacheListNameScript, luaGetCacheListNameKeys,luaGetCacheListNameArgs.toArray());
			if(cacheQueueNameList ==null || cacheQueueNameList.isEmpty()){
				return;
			}
			cacheQueueLengthList = 	AppStaticConfigs.scriptOpts3.getListLong(cacheQueueLengthScript, cacheQueueLengthKeys, cacheQueueNameList.toArray());
			List<String> priQueueNameList = QueueListEntity.cacheQueueNames2SendQueueNames(cacheQueueNameList);
			priQueueLengthList =  AppStaticConfigs.scriptOpts3.getListLong(cacheQueueLengthScript, priQueueLengthKeys, priQueueNameList.toArray());
			
			if(cacheQueueLengthList ==null || priQueueLengthList == null ||cacheQueueLengthList.size() != priQueueLengthList.size()){
				logger.error("获取缓存队列和发送队列长度时, redis 数据不一致!");
				return;
			}
			
				
			final QueueListEntity queueListEntity = new QueueListEntity(cacheQueueNameList, cacheQueueLengthList, priQueueLengthList);
			//1. 队列排序并按照通道进行分组
//			Map<String,ArrayList<String>> cacheQueueMap = gateSort(cacheQueueNameList);
			Map<String,List<String>> cacheQueueMap = queueListEntity.getCacheNameListMap();
			List<String> gateLinkList = new ArrayList<String>();
			if(cacheQueueMap != null && cacheQueueMap.size() > 0){
				Iterator<String>  gateLinkIdIterator = cacheQueueMap.keySet().iterator();
				while(gateLinkIdIterator.hasNext()){
					final String gateLinkId = gateLinkIdIterator.next();
					List<String> queueList = cacheQueueMap.get(gateLinkId);
					if(queueList != null && queueList.size() > 0){
						gateLinkList.add(gateLinkId);
						blockingPool.execute(new Runnable() {
							
							@Override
							public void run() {
								handlerQueue(gateLinkId,queueListEntity);
							}
						});
						handlerNum ++;
					}	
				}
			}
			logger.info("缓存队列处理通道  " + gateLinkList );
		} catch (Exception e) {
			logger.error("缓存队列处理过程发生错误:" + e.getMessage());
		}
	}
	
	
	private void handlerQueue(String gateLinkId,QueueListEntity queueListEntity) {
		//获取通道条数限制数,数据
		Long maxFlow = AppStaticConfigs.gateMaxFlow;
		Long cacheSize = AppStaticConfigs.cacheQueueProcessingNum;
		
		GateDef gateDef = AppStaticConfigs.configGateDefMap.get(gateLinkId);
		Integer threadNum = AppStaticConfigs.configGateLinkThreadNumMap.containsKey(gateLinkId) 
				? AppStaticConfigs.configGateLinkThreadNumMap.get(gateLinkId) : 1;
		if(gateDef !=  null ){
			Integer gateDefMaxFlow = gateDef.getMaxFlow();
			if(gateDefMaxFlow != null && gateDefMaxFlow > 0){
				maxFlow = AppStaticConfigs.sendQueueMaxProcess*maxFlow*threadNum;
			}
		}
		
		for(String queueName : queueListEntity.getCacheNameListMap().get(gateLinkId)){
			boolean handlerStatus = true;
			String priQueueName = QueueListEntity.changeSendQueueForCache(queueName);
			//缓存队列长度
			Long cacheQueueSize = queueListEntity.getCacheNameLengthMap().containsKey(queueName)
					? queueListEntity.getCacheNameLengthMap().get(queueName) 
					: 0l;
			//发送队列长度
			Long priQueueSize =  queueListEntity.getPriNameLengthMap().containsKey(priQueueName)
					? queueListEntity.getPriNameLengthMap().get(priQueueName) 
					: 0l;
			if(cacheQueueSize <= 0){
				continue;
			}
			do {
				if(maxFlow == 0){
					logger.error("通道配置错误,queueName: ["+queueName+"]");
					handlerStatus = false;
					break;
				}else if(priQueueSize < maxFlow){
					Long sendSize = maxFlow - priQueueSize;
					if(sendSize > cacheSize){
						sendSize = cacheSize;
					} 
					if(sendSize > cacheQueueSize){
						sendSize = cacheQueueSize;
					}
					priQueueSize += sendSize;
					cacheQueueSize -= sendSize;
					//发送短信
					try {
						pushToPriQueue(queueName,sendSize);
					} catch (Exception e) {
						e.printStackTrace();
						logger.error("缓存队列处理异常,message: "+e.getMessage());
					} 
					if(sendSize < cacheSize) {
						//已经全部发送完毕,跳出循环
						break;
					}
				} else {
					//因为按优先级排序了, 低优先级的也不用处理了
					handlerStatus = false;
					break;
				}
				cacheSize -= cacheQueueSize;
			} while (priQueueSize < maxFlow && handlerStatus);
			
			if ( !handlerStatus ){//结束后续队列的处理
				break;
			}
		}
	}
	
	
	private void pushToPriQueue(final String queueName, Long cacheSize) throws InterruptedException, BrokenBarrierException {
		 
		//获取缓存短信列表
		Date sdealTime = new Date();
		
		List<String> keys = new ArrayList<>();
		keys.add(queueName);
		keys.add(AppStaticConfigs.pSMCacheHash);
		List<List<String>> entities = AppStaticConfigs.scriptOpts3.getSMCacheAndHash(luaGetCacheQueueSmsScript, keys,0 + "", String.valueOf(cacheSize));
		List<SmsEntityForMt> lists = new ArrayList<>();
		for (int i = 0 ; i < entities.size() ; i ++) {
			List<String> list = entities.get(i);
			SmsEntityForMt smsEntity = SmsEntityForMt.parse(list);
			lists.add(smsEntity);
		}
		
		//过滤短信
		if (!lists.isEmpty()){
			int waitCount = 1;
			if(AppStaticConfigs.forbiddenTimeFilterOnOff){
				waitCount++;
				
			}
			CyclicBarrier barrier = new CyclicBarrier(waitCount);
			if(AppStaticConfigs.forbiddenTimeFilterOnOff){
				new ForbidTimeFilterTask(lists, barrier).start();
			}
			barrier.await();
			Date edealTime = new Date();
			//发送优先级队列
			List<byte[]> sendSerialnos = new ArrayList<byte[]>();
			List<byte[]> forbidTimeSerialnos = new ArrayList<byte[]>();
			// 禁发时段短信
			List<SmsEntityForMt> forbidTimeEntities = new ArrayList<>();
			for (SmsEntityForMt entity:lists) {
				String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();						
				if(entity.isIsforbidsms()){
					entity.setIsforbidsms(false);
					forbidTimeEntities.add(entity);
					forbidTimeSerialnos.add(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum));
				} else {
					sendSerialnos.add(AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
				}
			}
			
			//写入发送队列
			
			AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands(){

				@Override
				public void excute(RedisConnection connection) {
					
					
					//加入发送队列
					String sendQueueName = QueueListEntity.changeSendQueueForCache(queueName);
					if(! sendSerialnos.isEmpty()){
						byte[] sendListNameB = AppStaticConfigs.stringRedisSerializer.serialize(sendQueueName);
						connection.rPush(sendListNameB, sendSerialnos.toArray(new byte[1][]));
						
						//队列名加入相应列表中
						connection.sAdd(AppStaticConfigs.iSMPriListNamesSetB, sendListNameB);
						
						StringBuilder sb = new StringBuilder();
						for(int i = 0; i < sendSerialnos.size(); i++){
							if(i > 0){
								sb.append(",");
							}
							sb.append(AppStaticConfigs.stringRedisSerializer.deserialize(sendSerialnos.get(i)));
						}
						logger.mtprocc("[" + sb.toString() + "] 加入到发送队列 " + sendQueueName);
					}
					//清除发送短信数据结构
					if(!forbidTimeSerialnos.isEmpty()){
						connection.del(forbidTimeSerialnos.toArray(new byte[1][]));
						StringBuilder sb = new StringBuilder();
						for(int i = 0; i < forbidTimeSerialnos.size(); i++){
							if(i > 0){
								sb.append(",");
							}
							sb.append(AppStaticConfigs.stringRedisSerializer.deserialize(forbidTimeSerialnos.get(i)));
						}
						logger.mtprocc("[" + sb.toString() + "] 清除KEY " + sendQueueName);
					}
				}	
			});

			
			
			
			if(!forbidTimeEntities.isEmpty()){
				// 禁发时段短信,保存到禁发队列
				AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {
					
					@Override
					public void excute(RedisConnection connection) {
						
						for (SmsEntityForMt entity:forbidTimeEntities) {
							
							String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
							
							Map<byte[], byte[]> map = entity.getMapB();
	
							map.put(AppCanstant.SmsEntity.SdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(sdealTime)));//开始处理时间
							map.put(AppCanstant.SmsEntity.EdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(edealTime)));//处理完成时间
							
							map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_1B);
							map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_1_DESCB);
							
							connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
							
							connection.zAdd(AppStaticConfigs.iSMForbidZSetB, entity.getForbidTimeEnd(), AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
							
							logger.mtwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "] 重新加入禁发时段.");
						}
					}
				});
			}
		}
		
		
	}
	
	


	//缓存列表分类并排序
	private Map<String, ArrayList<String>> gateSort(
			List<String> cacheQueueList) {//[SM:GATEID:CACHE:LEVEL:LIST]
		Map<String, ArrayList<String>> map = new HashMap<String, ArrayList<String>>();
		for (String queueName : cacheQueueList){
			List<String> queueNames = StringUtils.splitToList(queueName, ":");
			String gateLinkId =   queueNames.get(1);
			ArrayList<String> queueNameList = null;
			if (map.containsKey(gateLinkId)){
				queueNameList = map.get(gateLinkId);
			} else {			
				queueNameList = new ArrayList<String>();
				map.put(gateLinkId, queueNameList);	
			}
		}
		for(String key : map.keySet()){
			ArrayList<String> queueNameList = map.get(key);
			queueNameList.sort(new Comparator<String>() {

				@Override
				public int compare(String o1, String o2) {
					
					List<String> o1List = StringUtils.splitToList(o1, ":");
					List<String> o2List = StringUtils.splitToList(o2, ":");					
					return StringUtils.toInteger(o2List.get(3)) - StringUtils.toInteger(o1List.get(3));
				}
			});
		}
		return map;
	}



	public void shutdown() {
		if (blockingPool != null) {
			blockingPool.shutdown();
		}
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
		this.interrupt();
	}


	
	private static class QueueListEntity {
		private List<String> cacheNameList;
		private List<String> priNameList;
		private Map<String,List<String>> cacheNameListMap; 
		private Map<String,List<String>> priNameListMap; 
		
		private Map<String,Long> cacheNameLengthMap; 
		private Map<String,Long> priNameLengthMap;
		
		public QueueListEntity(List<String> cacheNameList,List<Long> cacheNameLengthList,List<Long> priNameLengthList){
			this.cacheNameList = cacheNameList;
			priNameList = new ArrayList<String>();
			cacheNameListMap = new HashMap<String, List<String>>();
			priNameListMap = new HashMap<String, List<String>>();
			cacheNameLengthMap = new HashMap<String, Long>();
			priNameLengthMap = new HashMap<String, Long>();		
					
			for(int i = 0; i < this.cacheNameList.size(); i++){
				String cacheName = cacheNameList.get(i);
				List<String> cacheNames = StringUtils.splitToList(cacheName, ":");
				String gateLinkId =   cacheNames.get(1);
				String priName = changeSendQueueForCache(cacheName);
				priNameList.add(priName);
				cacheNameLengthMap.put(cacheName, cacheNameLengthList.get(i));
				priNameLengthMap.put(priName, priNameLengthList.get(i));
				List<String> cacheNameListTmp = null;
				List<String> priNameListTmp = null;
				if(cacheNameListMap.containsKey(gateLinkId)){
					cacheNameListTmp = cacheNameListMap.get(gateLinkId);
				} else {
					cacheNameListTmp = new ArrayList<String>();
					cacheNameListMap.put(gateLinkId, cacheNameListTmp);
				}
				if(priNameListMap.containsKey(gateLinkId)){
					priNameListTmp = priNameListMap.get(gateLinkId);
				} else {
					priNameListTmp = new ArrayList<String>();
					priNameListMap.put(gateLinkId, priNameListTmp);
				}
				cacheNameListTmp.add(cacheName);
				priNameListTmp.add(priName);
				
			}
			//按gateID优先级进行
			for(String key : cacheNameListMap.keySet()){
				List<String> queueNameList = cacheNameListMap.get(key);
				queueNameList.sort(new Comparator<String>() {

					@Override
					public int compare(String o1, String o2) {
						
						List<String> o1List = StringUtils.splitToList(o1, ":");
						List<String> o2List = StringUtils.splitToList(o2, ":");					
						return StringUtils.toInteger(o2List.get(3)) - StringUtils.toInteger(o1List.get(3));
					}
				});
			}
			for(String key : priNameListMap.keySet()){
				List<String> queueNameList = priNameListMap.get(key);
				queueNameList.sort(new Comparator<String>() {

					@Override
					public int compare(String o1, String o2) {
						
						List<String> o1List = StringUtils.splitToList(o1, ":");
						List<String> o2List = StringUtils.splitToList(o2, ":");					
						return StringUtils.toInteger(o2List.get(3)) - StringUtils.toInteger(o1List.get(3));
					}
				});
			}
			
			
		}
	
		
		public Map<String, List<String>> getCacheNameListMap() {
			return cacheNameListMap;
		}
		
		public Map<String, Long> getCacheNameLengthMap() {
			return cacheNameLengthMap;
		}
		
		public Map<String, Long> getPriNameLengthMap() {
			return priNameLengthMap;
		}
		public static List<String> cacheQueueNames2SendQueueNames(List<String> sendQueueNames){
			List<String> sendQueueList = new ArrayList<String>();
			for(String name : sendQueueNames){
				sendQueueList.add(changeSendQueueForCache(name));
			}
			return sendQueueList;
		}


		public static String changeSendQueueForCache(String queueName) {
			return queueName.replaceAll("Cache", "Pri");
		}
		
	}
	
	
}
